package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

public class RelativeStrengthIndexReducer extends Reducer<CodeDateKey, TimestampPriceDataPoint, Text, Text> {

	static long DAY_IN_MS = 24 * 60 * 60 * 1000;

	@Override
	public void reduce(CodeDateKey key, Iterable<TimestampPriceDataPoint> values,
			Context context) throws IOException, InterruptedException {

		TimestampPriceDataPoint nextPoint;
		
		int windowSizeInDays = context.getConfiguration().getInt("windowSize", CONFIG.WINDOW_SIZE);
		int windowStepSizeInDays = context.getConfiguration().getInt("windowStepSize", 1);

		long windowSizeInMS = windowSizeInDays * DAY_IN_MS;
		long windowStepSizeInMS = windowStepSizeInDays * DAY_IN_MS;

		Text outKey = new Text();
		Text outVal = new Text();

		SlidingWindow slidingWindow = new SlidingWindow(windowSizeInMS,
				windowStepSizeInMS, DAY_IN_MS);

		Iterator<TimestampPriceDataPoint> valueIterator = values.iterator();
		while (valueIterator.hasNext()) {

			while (!slidingWindow.WindowIsFull() && valueIterator.hasNext()) {

				nextPoint = valueIterator.next();

				TimestampPriceDataPoint pointCopy = new TimestampPriceDataPoint();
				pointCopy.copy(nextPoint);

				try {
					slidingWindow.AddPoint(pointCopy);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}

			if (slidingWindow.WindowIsFull()) {

				LinkedList<TimestampPriceDataPoint> window = slidingWindow
				.GetCurrentWindow();
			
				int countGain = 0;
				int countLoss = 0;
				double totalGain = 0;
				double totalLoss = 0;
				

				for(TimestampPriceDataPoint dataPoint: window)
				{
					if(dataPoint.getClosePrice()>dataPoint.getOpenPrice())
					{
						totalGain+=dataPoint.getClosePrice()-dataPoint.getOpenPrice();
						countGain++;
					} else{
						totalLoss += dataPoint.getOpenPrice()-dataPoint.getClosePrice();
						countLoss++;
					}
				}
				double averageGain = totalGain/countGain;
				double averageLoss = totalLoss/countLoss;
				double relativeStrength =  averageGain/averageLoss;
				double relativeStrengthIndex = 100 - 100/(1+relativeStrength);
				if (Double.isNaN(relativeStrengthIndex)){
					relativeStrengthIndex = 0;
				}

				String strBackDate = window.getLast().getDate();
				
				/*
				outKey.set("Code: " + key.getCode() + ", Date: "
						+ strBackDate);
				outVal.set("RSI: " + relativeStrengthIndex);
*/
				outKey.set(key.getCode() +  "\t" + strBackDate);
				outVal.set(""+relativeStrengthIndex);
				context.write(outKey, outVal);

				//step window forward

				slidingWindow.slideWindowForward();

			}

		} // end while

	} // end reduce

}
